package com.hex.ds.hdrs.app.service;

import com.hex.ds.hdrs.agent.po.AgentConf;
import com.hex.ds.hdrs.agent.service.AgentManager;
import com.hex.ds.hdrs.agent.service.AgentSession;
import com.hex.ds.hdrs.app.po.AppConf;
import com.hex.ds.hdrs.app.po.AppExecStrategyPo;
import com.hex.ds.hdrs.common.constant.Const;
import com.hex.ds.hdrs.common.constant.ErrorCode;
import com.hex.ds.hdrs.common.exception.BizException;
import com.hex.ds.hdrs.common.sort.*;
import com.hex.ds.hdrs.common.utils.HdrsDateUtil;
import com.hex.ds.hdrs.common.utils.ListUtil;
import com.hex.ds.hdrs.cron.po.CronTask;
import com.hex.ds.hdrs.log.service.LogService;
import com.hex.ds.hdrs.period.job.po.JobTask;
import com.hex.ds.hdrs.period.task.po.PeriodTask;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Package: com.hex.ds.hdrs.engine.po
 * @ClassName AppSession
 * @Description: App 内存信息存储
 * @Author: jd.chen
 * @Date 2021/7/5 17:56
 * @Version v1.0
 **/
@Slf4j
@Data
public class AppSession {

    // periodRunningList 同步锁
    private final byte[] periodListSyn = {};
    // cronRunningList 同步锁
    private final byte[] cronListSyn = {};

    private static final String RES = "res";
    private static final String CONCURRENT = "concurrent";

    //是否在资源重新计算中
    private String mark = Const.APP_MARK_STOP;

    // 应用编码
    private String appCode;
    // 应用活动状态
    private String appStatus;
    // 应用配置
    private AppConf appConf;
    // 应用管理
    private AppManager appManager;
    // 当前占用资源数
    private final AtomicInteger curResNum = new AtomicInteger(0);
    // 当前占用并发数
    private final AtomicInteger curConcurrentNum = new AtomicInteger(0);
    // 应用分发策略
    private List<AppExecStrategyPo> execStrategyList;
    // 排序策略
    private String sortStrategy;
    // 任务互斥池
    private List<String> mutexTaskCodeList;

    // 获取应用权重策略
    private String weightType;
    // 应用选择权重
    private int weight;
    // 应用选择权重-计算时间
    private long weightUpTs = 0;

    // 节点配置列表
    private List<AgentConf> agentConfList;
    // 节点数
    private int agentSize = 0;
    // 有效的节点
    private List<AgentSession> agentSessionList = new CopyOnWriteArrayList<>();
    // 各节点占权重 <agentCode,weightNum>
    private Map<String, Integer> agentWeightMap;

    // 正在运行队列(周期)
    private List<PeriodTask> periodRunningList = Collections.synchronizedList(new ArrayList<>());
    // 正在运行队列(定时)
    private List<CronTask> cronRunningList = Collections.synchronizedList(new ArrayList<>());
    // 重新计算资源时临时运行队列(周期)
    private List<PeriodTask> deltaList = Collections.synchronizedList(new ArrayList<>());

    // 当前占用资源数  上次全量计算时间
    private long lstComputeTs;
    // 当前信息 <agentCode,CurInfo>
    //private Map<String, CurInfo> curMap = new ConcurrentHashMap<>();

    // 上次资源不足
    private long lstOverResTs = 0;
    // 连续资源不足次数
    private final AtomicInteger overResCount = new AtomicInteger(0);

    // 上次并发不足
    private long lstOverConcurrentTs = 0;
    // 连续并发不足次数
    private final AtomicInteger overConcurrentCount = new AtomicInteger(0);

    private LogService logService;

    public AppSession(AppConf appConf, List<AgentConf> agentConfList, String weightType,
                      Map<String, Integer> agentWeightMap, AppManager appManager) {
        this.appCode = appConf.getAppCode();
        this.appStatus = appConf.getAppSatus();
        this.appConf = appConf;
        this.appManager = appManager;
        this.agentConfList = agentConfList;
        this.execStrategyList = appConf.getExecStrategyList();
        this.sortStrategy = appConf.getDispatcherStrategyCode();
        this.mutexTaskCodeList = appConf.getMutexPool();
        this.agentSize = this.agentConfList.size();
        this.weightType = weightType;
        this.agentWeightMap = agentWeightMap;
        this.logService = appManager.getLogService();
    }

    /**
     * 验证 appsession
     *
     * @param task
     * @return
     */
    public boolean checkRes(PeriodTask task) {
        log.debug("验证 appsession【AgentSession.checkRes】开始=======》");
        AppExecStrategyPo strategy = this.getAppStrategy();

        // 任务资源数
        int taskResNum = task.getResNum();
        // 获取当前应用资源数
        int limitResNum = this.getResNum(strategy) - taskResNum;
        // 判断资源数是否满足
        if (this.curResNum.get() > limitResNum) {
            log.warn("{}【资源数不足】【checkRes】taskResNum:{},curResNum:{},limitRes:{}", appCode, taskResNum, curResNum.get(), this.getResNum(strategy));
            overResCount.addAndGet(1);
            if (needAddLog(RES)) {
                logService.resNotEnoughAppLog(appConf, "资源数达到最大!");
            }
            return false;
        } else {
            overResCount.set(0);
        }

        // 获取当前应用并发数
        int limitConcurrentNum = this.getConcurrentNum(strategy) - 1;
        // 判断并发数是否满足
        if (this.curConcurrentNum.get() > limitConcurrentNum) {
            log.warn("【{}】【并发数不足】【checkRes】curConcurrentNum:{},limitConcurrentNum:{}", appCode, curConcurrentNum.get(), this.getConcurrentNum(strategy));
            overConcurrentCount.addAndGet(1);
            if (needAddLog(CONCURRENT)) {
                logService.resNotEnoughAppLog(appConf, "并发数达到最大!");
            }
            return false;
        } else {
            overConcurrentCount.set(0);
        }

        // addTaskSession(task);
        return true;
    }

    public boolean needAddLog(String type) {
        if (RES.equals(type)) {
            if (overResCount.get() > 10 && HdrsDateUtil.getMillis(new Date()) - lstOverResTs > 300000) {
                lstOverResTs = HdrsDateUtil.getMillis(new Date());
                return true;
            } else {
                return false;
            }
        } else if (CONCURRENT.equals(type)) {
            if (overConcurrentCount.get() > 10 && HdrsDateUtil.getMillis(new Date()) - lstOverConcurrentTs > 300000) {
                lstOverConcurrentTs = HdrsDateUtil.getMillis(new Date());
                return true;
            } else {
                return false;
            }
        } else {
            log.error("【needAddLog】类型异常:{}", type);
            return false;
        }
    }

    /**
     * 获取资源数
     *
     * @return
     */
    public int getResNum(AppExecStrategyPo strategy) {
        log.debug("获取资源数【AppSession.getResNum】开始==========》");
        if (null == strategy) {
            return appConf.getResNum();
        } else {
            return strategy.getResNum();
        }
    }

    /**
     * 获取并发数
     *
     * @return
     */
    public int getConcurrentNum(AppExecStrategyPo strategy) {
        log.debug("获取并发数【AppSession.getConcurrentNum】开始==========》");
        if (null == strategy) {
            return appConf.getConcurrentNum();
        } else {
            return strategy.getConcurrentNum();
        }
    }

    /**
     * 获取应用策略
     *
     * @return
     */
    public AppExecStrategyPo getAppStrategy() {
        log.debug("获取应用策略【AppSession.getAppStrategy】开始==========》");
        if (!ListUtil.isEmpty(this.execStrategyList)) {
            int hour = HdrsDateUtil.getHour();
            for (AppExecStrategyPo strategy : execStrategyList) {
                int stHour = Integer.parseInt(strategy.getStHour().split(":")[0]);
                int endHour = Integer.parseInt(strategy.getEndHour().split(":")[0]);
                if (hour >= stHour && hour <= endHour) {
                    return strategy;
                }
            }
        }
        return null;
    }

    /**
     * 更新session 周期任务变化
     *
     * @param task
     */
    public void addTaskSession(PeriodTask task) {
        log.debug("更新session 新增周期任务【AgentSession.addTaskSession】开始=======》");
        synchronized (periodListSyn) {
            if (this.getCurConcurrentNum().get() + 1 > this.getResNum(this.getAppStrategy())) {
                throw new BizException(ErrorCode.ERROR_820001);
            }
            List<PeriodTask> taskList = new ArrayList<>();
            taskList.add(task);
            if (this.periodRunningList.addAll(taskList)) {
                this.curResNum.addAndGet(task.getResNum());
                this.curConcurrentNum.addAndGet(1);
            }
        }
    }

    /**
     * 更新session 有效的agent
     *
     * @param agentSession
     */
    public void addAgentSession(AgentSession agentSession) {
        log.debug("更新session 有效的agent【AppSession.addAgentSession】开始=======》");
        if (agentSession == null) {
            return;
        } else {
            this.agentSessionList.add(agentSession);
        }
    }

    /**
     * 更新session 定时任务变化
     *
     * @param cronTask
     */
    public void addTaskSession(CronTask cronTask) {
        log.debug("更新session 新增定时任务【AgentSession.addTaskSession】开始=======》");
        synchronized (cronListSyn) {
            List<CronTask> taskList = new ArrayList<>();
            taskList.add(cronTask);
            this.cronRunningList.addAll(taskList);
        }
    }


    /**
     * 更新session 周期任务变化
     *
     * @param task
     */
    public void delTaskSession(PeriodTask task) {
        log.debug("更新session 删除周期任务【AgentSession.delTaskSession】开始=======》");
        try {
            //判断是否在资源重新计算中
            if (this.mark.equals(Const.APP_MARK_STOP)) {
                synchronized (periodListSyn) {
                    List<PeriodTask> taskList = new ArrayList<>();
                    taskList.add(task);
                    if (this.periodRunningList.removeAll(taskList)) {
                        this.curResNum.addAndGet(-task.getResNum());
                        this.curConcurrentNum.addAndGet(-1);
                    }
                }
            } else {
                List<PeriodTask> taskList = new ArrayList<>();
                taskList.add(task);
                this.periodRunningList.removeAll(taskList);
                this.deltaList.addAll(taskList);
            }
        } catch (Exception e) {
            log.error("【delTaskSession】异常:{}", e.getMessage(), e);
        }
    }

    /**
     * 更新session 周期任务变化
     *
     * @param taskCode
     */
    public void delTaskSession(String taskCode) {
        log.info("更新session 删除周期任务【AgentSession.delTaskSession】taskCode: {}", taskCode);
        try {
            for (int i = 0; i < this.periodRunningList.size(); i++) {
                PeriodTask periodTask = this.periodRunningList.get(i);
                if (periodTask != null && taskCode.equals(periodTask.getTaskCode())) {
                    this.delTaskSession(periodTask);
                    return;
                }
            }
        } catch (Exception e) {
            log.error("【{}】【delTaskSession】异常:{}", taskCode, e.getMessage(), e);
        }
    }

    /**
     * 更新session 定时任务变化
     *
     * @param cronTask
     */
    public void delTaskSession(CronTask cronTask) {
        log.debug("更新session 删除定时任务【AgentSession.delTaskSession】开始=======》");
        synchronized (cronListSyn) {
            List<CronTask> taskList = new ArrayList<>();
            taskList.add(cronTask);
            this.cronRunningList.removeAll(taskList);
        }
    }

    /**
     * 节点资源判断是否充足
     *
     * @return
     */
    public AgentSession getFitAgent(AgentManager agentManager, JobTask jobTask, PeriodTask task) {
        log.debug("节点资源判断是否充足【AppSession.getFitAgent】=======》");

        //1. 判断app下是否有相应agent
        if (ListUtil.isEmpty(agentSessionList)) {
            log.debug("【AppSession.getFitAgent】【警告】agentSession 不活动……………………");
            throw new BizException(ErrorCode.ERROR_820002);
        }

        // 2. 单节点下筛选有效合理的节点
        try {
            AgentSession agentSession = this.oneAgentDeal(agentManager, jobTask, task);
            if (agentSession != null) {
                // agentSession.addTaskSession(task);
                return agentSession;
            }
        } catch (BizException b) {
            log.debug("oneAgentDeal 失败，尝试使用 moreAgentDeal 获取 agent");
        }

        // 3. 多节点下筛选有效合理的节点
        return this.moreAgentDeal(task);
    }

    /**
     * 单节点下筛选有效合理的节点
     *
     * @param agentManager
     * @param jobTask
     * @param task
     * @return
     */
    public AgentSession oneAgentDeal(AgentManager agentManager, JobTask jobTask, PeriodTask task) {
        log.debug("单节点下筛选有效合理的节点【AppSession.oneAgentDeal】=======》");
        String agentCode;
        AgentSession agentSession;

        //1. 是否指定节点
        if (!StringUtils.isEmpty(task.getAppointAgentCode())) {
            agentCode = task.getAppointAgentCode();
            log.debug("指定节点执行【AppSession.getFitAgent】执行节点：{}", agentCode);
            //验证节点，满足则返回节点编码
            agentSession = agentManager.getAgentSession(agentCode);
            if (agentSession.checkRes(task)) {
                return agentSession;
            } else {
                log.warn("【{}】指定节点执行【AppSession.getFitAgent】资源不足", agentCode);
                throw new BizException(ErrorCode.ERROR_820003);
            }
        }

        //2. 同批次同节点
        if (!StringUtils.isEmpty(task.getIsOneAgent())) {
            PeriodTask upTask = jobTask.getUpTask(task);
            if (upTask != null) {
                agentCode = upTask.getAgentCode();
                log.debug("同批次同节点执行【AppSession.getFitAgent】执行节点：{}", agentCode);
                agentSession = agentManager.getAgentSession(agentCode);
                if (agentSession.checkRes(task)) {
                    return agentSession;
                } else {
                    log.warn("【{}】同批次同节点执行【AppSession.getFitAgent】资源不足", agentCode);
                    throw new BizException(ErrorCode.ERROR_820003);
                }
            }
        }

        //3. 获取节点列表
        if (agentSessionList.size() == 1) {
            agentCode = agentSessionList.get(0).getAgentCode();
            log.debug("节点执行【AppSession.getFitAgent】执行节点：{}", agentCode);
            agentSession = agentManager.getAgentSession(agentCode);
            if (agentSession.checkRes(task)) {
                return agentSession;
            } else {
                log.warn("【{}】节点执行【AppSession.getFitAgent】资源不足", agentCode);
                throw new BizException(ErrorCode.ERROR_820003);
            }
        }
        return null;
    }

    /**
     * 多节点下筛选有效合理的节点
     *
     * @param task
     * @return
     */
    public AgentSession moreAgentDeal(PeriodTask task) {
        log.debug("多节点下筛选有效合理的节点【AppSession.moreAgentDeal】=======》");
        if (ListUtil.isEmpty(agentSessionList)) {
            log.error("【moreAgentDeal】agentSessionList个数0");
            return null;
        }

        //1. 筛选出有效的节点
        List<AgentSession> validAgentList = new ArrayList<>();
        for (AgentSession agentSession : agentSessionList) {
            log.debug("【moreAgentDeal】agentCode: {}", agentSession.getAgentCode());
            if (!Const.AGENT_STATUS_STOP.equals(agentSession.getStatus())) {
                if (agentSession.checkRes(task)) {
                    validAgentList.add(agentSession);
                }
            }
        }

        int listSize = validAgentList.size();
        log.info("【moreAgentDeal】validAgentList.size(): {}", listSize);
        //2. 超过1个节点满足，根据策略选取
        if (listSize == 1) {
            log.debug("有效的节点【AppSession.moreAgentDeal】执行节点: {}", validAgentList.get(0).getAgentCode());
            AgentSession agentSession = validAgentList.get(0);
            return agentSession;
        } else if (listSize > 1) {
            AgentSession agentSession = this.chooseAgentByStrategy(validAgentList);
            log.debug("多节点下筛选有效的节点【AppSession.moreAgentDeal】执行节点: {}", agentSession.getAgentCode());
            return agentSession;
        } else {
            log.warn("【getAgentSession】agentSession 资源不足……………………");
            throw new BizException(ErrorCode.ERROR_820003);
        }
    }

    /**
     * 超过1个节点满足，根据权重选取
     *
     * @param agentSessionList
     * @return
     */
    public AgentSession chooseAgentByStrategy(List<AgentSession> agentSessionList) {
        log.debug("超过1个节点满足，根据权重选取【AppSession.chooseAgentByStrategy】=======》");
        int allNum = 0;
        for (AgentSession agentSession : agentSessionList) {
            allNum += agentWeightMap.get(agentSession.getAgentCode());
        }

        int rNum = RandomUtils.nextInt(1, allNum);
        int res = 0;
        for (AgentSession agentSession : agentSessionList) {
            res += agentWeightMap.get(agentSession.getAgentCode());
            if (rNum <= res) {
                return agentSession;
            }
        }
        log.error("【chooseAgentByStrategy】..........");
        return agentSessionList.get(0);
    }

    /**
     * 是否包含agent
     *
     * @param agentCode
     * @return
     */
    public boolean containAgent(String agentCode) {
        for (AgentConf agentConf : agentConfList) {
            if (agentCode.equals(agentConf.getAgentCode())) {
                return true;
            }
        }
        return false;
    }

    /**
     * 根据app策略选择排序方式
     *
     * @return
     */
    public SortByStrategy sortStrategy() {
        log.debug("根据app策略选择排序方式【AppSession.sortStrategy】=========》");
        switch (sortStrategy) {
            case Const.SORT_PRIORITY:
                return new SortByPriority();
            case Const.SORT_PRIORITY_RES_DESC:
                return new SortByPriorityAndResDesc();
            case Const.SORT_PRIORITY_RES_ASC:
                return new SortByPriorityAndResAsc();
            case Const.SORT_EARLIER_TS:
                return new SortByEarlierTs();
            default:
                throw new BizException(ErrorCode.ERROR_888000);
        }
    }

    /**
     * 获取应用权重
     *
     * @param weightUpInterval
     * @return
     */
    public int getAppWeight(int weightUpInterval) {
        log.debug("获取应用权重【AppSession.getAppWeight】=========》");
        long curTs = HdrsDateUtil.getMillis(new Date());
        if (curTs - weightUpTs <= weightUpInterval) {
            return weight;
        } else {
            weightUpTs = curTs;
            switch (weightType) {
                case Const.GROUP_APP_RES:
                    weight = appConf.getResNum();
                    return weight;
                case Const.GROUP_APP_CONCURRENT:
                    weight = appConf.getConcurrentNum();
                    return weight;
                case Const.GROUP_APP_STRATEGY_RES:
                    weight = this.getResNum(this.getAppStrategy());
                    return weight;
                case Const.GROUP_APP_STRATEGY_CONCURRENT:
                    weight = this.getConcurrentNum(this.getAppStrategy());
                    return weight;
                default:
                    throw new BizException(ErrorCode.ERROR_820008);
            }
        }
    }

    /**
     * 删除应用下的agentSession
     *
     * @param agentSession
     */
    public void delAgentSession(AgentSession agentSession) {
        agentSessionList.remove(agentSession);
    }

    public String getAppName() {
        return appConf.getAppName();
    }

    public boolean containTask(PeriodTask task) {
        return periodRunningList.contains(task);
    }

    /**
     * @Method: againAppResAndCurr <br>
     * @Param: [] <br>
     * @Return: void <br>
     * @Description：重新计算appSession资源数和并发数<br>
     * @Author： gc.zuo<br>
     * @Date： 2021/6/29 15:44 <br>
     * @Version： V1.0<br>
     */
    public void againAppResAndCurr() {
        log.debug("重新计算appSession资源数和并发数");
        try {
            synchronized (periodListSyn) {
                List<PeriodTask> taskList = this.periodRunningList;
                int resNumTotal = 0;
                //遍历计算资源数
                for (PeriodTask task : taskList) {
                    resNumTotal += task.getResNum();
                }
                //重新赋值
                int deltaResNumTotal = 0;
                for (PeriodTask task : this.deltaList) {
                    deltaResNumTotal += task.getResNum();
                }
                this.curResNum.set(resNumTotal - deltaResNumTotal);
                this.curConcurrentNum.set(taskList.size() - this.deltaList.size());
                this.deltaList.clear();
            }
        } catch (Exception e) {
            log.error("{}重新计算资源数并发数出错: {}", this.appCode, e.getMessage(), e);
        }
    }

    /**
     * @Method: setMutexTask <br>
     * @Param: [codeList] <br>
     * @Return: void <br>
     * @Description：互斥池重新赋值<br>
     * @Author： gc.zuo<br>
     * @Date： 2022/7/11 16:23 <br>
     * @Version： V1.0<br>
     */
    public void setMutexTask(List<String> codeList) {
        this.mutexTaskCodeList = codeList;
    }
}
